package com.lhczf.lucenedb.consumer;

import com.lhczf.lucenedb.bean.DataWrap;
import com.lhczf.lucenedb.bean.MergeDataConfig;
import com.lhczf.lucenedb.bean.SystemConfig;
import com.lhczf.lucenedb.service.LuceneDbServer;
import com.lhczf.lucenedb.util.BeanUtil;
import com.lhczf.lucenedb.util.FileUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.lucene.facet.taxonomy.directory.DirectoryTaxonomyWriter;
import org.apache.lucene.index.IndexWriter;
import org.apache.lucene.store.Directory;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * @author: 刘华春
 * @date: 2019/2/6
 */
@Slf4j
public class MergeDataConsumer extends AbstractDataConsumer {

    private Map<String, IndexWriter> allWriter = new HashMap<>(16);
    private static final String NAME = "mergeData";

    @Override
    AbstractDataConsumer createDataConsumer() {
        return new MergeDataConsumer();
    }

    @Override
    public String configIndexName() {
        return NAME;
    }

    @Override
    DataWrap takeDataWrap() {
        BlockingQueue blockingQueue = DATA_QUEUE.computeIfAbsent(MERGE_DATA_QUEUE_KEY, k -> new LinkedBlockingQueue());
        return getDataWrapFromQueue(blockingQueue);
    }

    @Override
    public void configTaxoIndexWriter(DirectoryTaxonomyWriter taxonomyWriter) {
        //不需要做什么操作，在这个方法里面
    }

    @Override
    public boolean autoRecover() {
        return true;
    }

    @Override
    public int runtimeThreadNum() {
        return BeanUtil.getInstance().getBean(MergeDataConfig.class).getRunThreadNum();
    }

    @Override
    public boolean hasConfigTaxoWriter() {
        return false;
    }

    @Override
    public void run() {
        log.info("索引合并线程已经启动。");
        LuceneDbServer server = BeanUtil.getInstance().getBean(LuceneDbServer.class);
        SystemConfig config = BeanUtil.getInstance().getBean(SystemConfig.class);

        while (true) {
            DataWrap dataWrap = takeDataWrap();
            if (dataWrap == null) {
                continue;
            }
            IndexWriter indexWriter = allWriter.get(dataWrap.getIndexName());
            if (indexWriter == null) {
                init(server, config, dataWrap);
            }
            String indexDir = dataWrap.getIndexDir();
            indexWriter = allWriter.get(dataWrap.getIndexName());
            Directory directory = server.makeDirectory(indexDir);
            try {
                indexWriter.addIndexes(directory);
                indexWriter.commit();
            } catch (IOException e) {
                log.error("索引合并出现异常，异常目录为：{}", server.findIndexPath(directory), e);
            } finally {
                try {
                    directory.close();
                } catch (IOException e) {
                    log.error("", e);
                }
            }
            log.info("索引合并结束，被合并的索引为：{}", server.findIndexPath(directory));
            //将已经合并完的索引目录存入BACKUP_INDEX_DATA_MAP
            //这个MAP的key为索引名称，value为当前索引已经合并完的索引目录的集合
            Map<String, List<String>> daysDirs = BACKUP_INDEX_DATA_MAP.computeIfAbsent(dataWrap.getIndexName(), k -> new HashMap<>(16));
            File file = new File(indexDir);
            file = FileUtil.findParentFile(file);
            if (file == null) {
                log.warn("目录结构异常。获取不到父目录的名称：{}", indexDir);
            } else {
                String dayName = file.getName();
                List<String> dayDirList = daysDirs.computeIfAbsent(dayName, k -> new ArrayList<>());
                dayDirList.add(indexDir);
            }
        }
    }

    private void init(LuceneDbServer server, SystemConfig config, DataWrap dataWrap) {
        IndexWriter indexWriter = allWriter.get(dataWrap.getIndexName());
        if (indexWriter == null) {
            StringBuilder sb = new StringBuilder(config.getIndexData())
                    .append(File.separator).append(SystemConfig.ALL_INDEXS_DIR)
                    .append(File.separator).append(SystemConfig.INDEX_DIR_NAME)
                    .append(File.separator).append(dataWrap.getIndexName());
            indexWriter = server.createIndexWriter(sb.toString(), dataWrap.getIndexName());
            log.info("创建indexWriter成功， 索引目录为： {}", sb.toString());
            allWriter.putIfAbsent(dataWrap.getIndexName(), indexWriter);
        }
    }
}
